package com.bridgeintelligent.tag.bulkload.service.core;

import com.bridgeintelligent.tag.bulkload.api.FileExportDto;
import com.bridgeintelligent.tag.bulkload.api.LookLikeExportDto;
import com.bridgeintelligent.tag.bulkload.api.fileload.FileLoadResult;
import com.bridgeintelligent.tag.bulkload.api.fileload.Filter;
import com.bridgeintelligent.tag.bulkload.api.fileload.FilterGroup;
import com.bridgeintelligent.tag.bulkload.service.core.fileload.ConditionBuilder;
import com.bridgeintelligent.tag.constants.bulkload.FileLoadMqMsg;
import com.bridgeintelligent.tag.constants.bulkload.MQConfig;
import com.bridgeintelligent.tag.mapper.EsTimeMapper;
import com.bridgeintelligent.tag.mapper.FileExportMapper;
import com.bridgeintelligent.tag.mapper.IssueTaskMapper;
import com.bridgeintelligent.tag.utils.DateHelper;
import com.bridgeintelligent.tag.utils.FieldHandler;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.AllArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

/**
 * @author JackCheung
 * @date 2020/4/26 2:32 PM.
 * description:客群导出文件mq代理
 */

@AllArgsConstructor
@Slf4j
@Component
@RocketMQMessageListener(topic = MQConfig.FILE_LOAD_TAG_TOPIC, consumerGroup = MQConfig.FILE_LOAD_TAG_CONSUMER_GROUP,consumeThreadMax=1)
public class FileLoadDelegate implements RocketMQListener<String> {

    private ObjectMapper objectMapper;

    private FileExportMapper fileExportMapper;

    private ConditionBuilder conditionBuilder;

    private FileLoadHandler fileLoadHandler;

    private FileLoadLookLikeHandler fileLoadLookLikeHandler;

    private static final DateTimeFormatter YYYY_MM = DateTimeFormatter.ofPattern("yyyyMM");

    private static final String SUCCESS_FLAG = "2";

    private IssueTaskMapper issueTaskMapper;

    private EsTimeMapper esTimeMapper;

    private static final String LAST_CLOSING_DATE = "last_closing_date";

    @SneakyThrows
    @Override
    public void onMessage(String message) {

        long time = System.currentTimeMillis();
        FileLoadMqMsg fileLoadMqMsg = objectMapper.readValue(message, FileLoadMqMsg.class);
        log.info("-------本地下载消费开始:"+fileLoadMqMsg.toString());
        try {
            Thread.sleep(1000);
            //幂等性控制，防止单次消费处理时间大于MQ默认15分钟触发重试机制
            int count = issueTaskMapper.transNum(fileLoadMqMsg.getUniqueIdentify());
            log.info("当前线程：{}，处理：{}，获取count:{}",Thread.currentThread().getName(),fileLoadMqMsg.getUniqueIdentify(),count);
            if (count > 0){
                log.error("任务：{}已经处理过",fileLoadMqMsg.getUniqueIdentify());
                return;
            }
            issueTaskMapper.addTransNum(fileLoadMqMsg.getUniqueIdentify());


            //查询导出信息根据lookConfigId字段判断是客群导出还是lookLike导出，分别走不同的逻辑处理
            FileExportDto fileExportDto = fileExportMapper.findFileExport(fileLoadMqMsg.getId());
            if (fileExportDto == null) {
                log.error("\ntable TAG_CUSTOMERS_EXPORT cannot found id=" + fileLoadMqMsg.getId() + ", check data\n");
                return;
            }
            //如果lookConfigId为空，说明是常规客群导出，处理逻辑不变，如果不为空说明是looklike导出走新逻辑
            if (StringUtils.isEmpty(fileExportDto.getLookConfigId())){

                FileExportDto dbDto = fileExportMapper.findIndexById(fileExportDto.getCustomerId());
                if (dbDto == null){
                    log.error("\ntable TAG_CUSTOMERS cannot found id=" + fileExportDto.getCustomerId() + ", check data\n");
                    return;
                }
                fileExportDto.setIndexId(dbDto.getIndexId());
                fileExportDto.setTimeType(dbDto.getTimeType());

                fileExportDto.setFieldValue(fileExportDto.getFieldValue()+","+LAST_CLOSING_DATE);
                //此处根据dataDate去校验ES是否正在写入数据
                if (!checkWrite(fileExportDto.getId(),fileExportDto.getDataDate(),fileExportDto.getIndexId())){
                    log.error("此时ES正在写入数据");
                    return;
                }
                List<Filter> filters = new ArrayList<>();
                FilterGroup filterGroup = conditionBuilder.findFilterGroup("-", fileExportDto.getCustomerId(), filters)
                        .get(0);
                FileLoadResult fileLoadResult = fileLoadHandler.handler(fileExportDto, filterGroup);
                fileExportMapper.updateFilePath(fileExportDto.getId(), fileLoadResult.getFilePath(), SUCCESS_FLAG, fetchDate(), String.valueOf(fileLoadResult.getNum()));

            }else{
                //根据id获取导出配置信息
                LookLikeExportDto lookLikeConfig = fileExportMapper.findLookLikeConfig(fileExportDto.getLookConfigId());
                FileLoadResult fileLoadResult = fileLoadLookLikeHandler.handler(fileExportDto, lookLikeConfig);
                fileExportMapper.updateFilePath(fileExportDto.getId(), fileLoadResult.getFilePath(), SUCCESS_FLAG, fetchDate(), String.valueOf(fileLoadResult.getNum()));
            }
        } catch (Exception e) {
            log.error("客群导出文件出现异常=======>", e);
            fileExportMapper.exportFail(fileLoadMqMsg.getId(),"3", DateHelper.currentDateTime(),"客群导出文件失败");
        } finally {
            log.debug("客群导出文件总耗时=====>{}ms", (System.currentTimeMillis() - time));
        }
    }

    private boolean checkWrite(String exportId,String dataDate,String indexId){
        boolean isCheck = true;
        if (!StringUtils.isEmpty(dataDate)){
            //说明是日更操作
            if (FieldHandler.pattern.matcher(dataDate).matches() && dataDate.length() == FieldHandler.LENGTH8){
                int count = issueTaskMapper.batchCount(1,dataDate,indexId);
                if (count>0){
                    fileExportMapper.exportFail(exportId,"3", DateHelper.currentDateTime(),"ES正在写入数据");
                    isCheck = false;
                }
            }else{
                String esSymbol = Optional.ofNullable(esTimeMapper.findEsSymbol(indexId,dataDate)).orElse(dataDate);
                int count = issueTaskMapper.batchCount(2,esSymbol,indexId);
                if (count>0){
                    fileExportMapper.exportFail(exportId,"3", DateHelper.currentDateTime(),"ES正在写入数据");
                    isCheck = false;
                }
            }
        }
        return isCheck;
    }

    private static String fetchDate() {
        return LocalDateTime.now()
                            .format(YYYY_MM);
    }
}
